package com.thinkingcoder.tcmq.client;

import org.springframework.util.LinkedMultiValueMap;
import org.springframework.util.MultiValueMap;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.TypeReference;
import com.thinkingcoder.common.util.HttpUtils;
import com.thinkingcoder.common.util.ThreadUtils;
import com.thinkingcoder.tcmq.core.model.Message;
import com.thinkingcoder.tcmq.core.model.Result;
import com.thinkingcoder.tcmq.core.model.Stat;

import lombok.Getter;

/**
 * broker for topics.
 */
public class TcBroker {

    private static final String defaultBrokerServer = "localhost:8765/tcmq";

    private String brokerServer;

    public TcBroker() {
        this.brokerServer = defaultBrokerServer;
    }

    public TcBroker(String brokerServer) {
        this.brokerServer = brokerServer;
    }

    @Getter
    private static TcBroker Default = new TcBroker();

    @Getter
    private MultiValueMap<String, TcConsumer<?>> consumers = new LinkedMultiValueMap<>();

    public void addConsumer(String topic, TcConsumer<?> consumer) {
        consumers.add(topic, consumer);
    }

    static {
        // init();
    }

    public static void init() {
        TcBroker.init(getDefault());
    }

    public static void init(TcBroker broker) {
        ThreadUtils.getDefault().init(1);
        ThreadUtils.getDefault().schedule(() -> {
            MultiValueMap<String, TcConsumer<?>> multConsumers = broker.getConsumers();
            multConsumers.forEach((topic, tempConsumers) -> {
                tempConsumers.forEach(tempConsumer -> {
                    Message<?> receiveMsg = tempConsumer.receive(topic);
                    if (receiveMsg == null) {
                        return;
                    }
                    try {
                        tempConsumer.getListener().onMessage(receiveMsg);
                        tempConsumer.ack(topic, receiveMsg);
                    } catch (Exception e) {
                        // TODO
                    }
                });
            });
        }, 100, 100);
    }

    public TcProducer createProducer() {
        return new TcProducer(this);
    }

    public TcConsumer<?> createConsumer(String topic) {
        TcConsumer<?> consumer = new TcConsumer<>(this);
        consumer.subscribe(topic);
        return consumer;
    }

    public void autoCreateTopic(String topic, String id, boolean allowAutoCreate) {
        System.out
            .println(" ===>>> autoCreateTopic topic/id/allowAutoCreate: " + topic + "/" + id + "/" + allowAutoCreate);

        Result<String> result =
            HttpUtils.httpGet("http://" + brokerServer + "/autoCreateTopic?t=" + topic + "&aac=" + allowAutoCreate,
                new TypeReference<>() {});

        System.out.println(" ===>>> autoCreateTopic result: " + result);
    }

    public boolean send(String topic, Message message) {
        System.out.println(" ===>>> send topic/message: " + topic + "/" + message);
        System.out.println(JSON.toJSONString(message));

        Result<String> result = HttpUtils.httpPost("http://" + brokerServer + "/send?t=" + topic,
            JSON.toJSONString(message), new TypeReference<>() {});
        System.out.println(" ===>>> send result: " + result);

        return result.getCode() == 1;
    }

    public void subscribe(String topic, String consumerId) {
        System.out.println(" ===>>> subscribe topic/consumerId: " + topic + "/" + consumerId);
        Result<String> result = HttpUtils.httpGet(
            "http://" + brokerServer + "/subscribe?t=" + topic + "&cid=" + consumerId, new TypeReference<>() {});
        System.out.println(" ===>>> subscribe result: " + result);
    }

    public void unSubscribe(String topic, String consumerId) {
        System.out.println(" ===>>> unSubscribe topic/consumerId: " + topic + "/" + consumerId);

        Result<String> result = HttpUtils.httpGet(
            "http://" + brokerServer + "/unSubscribe?t=" + topic + "&cid=" + consumerId, new TypeReference<>() {});

        System.out.println(" ===>>> unSubscribe result: " + result);
    }

    public <T> Message<T> receive(String topic, String consumerId) {
        System.out.println(" ===>>> receive topic/consumerId: " + topic + "/" + consumerId);

        Result<Message<String>> result = HttpUtils
            .httpGet("http://" + brokerServer + "/receive?t=" + topic + "&cid=" + consumerId, new TypeReference<>() {});
        System.out.println(" ===>>> receive result: " + result);

        return result.getData() == null ? null : (Message<T>)result.getData();
    }

    public boolean ack(String topic, String cid, int readOffset) {
        System.out.println(" ===>>> ack topic/cid/readOffset: " + topic + "/" + cid + "/" + readOffset);

        Result<String> result =
            HttpUtils.httpGet("http://" + brokerServer + "/ack?t=" + topic + "&cid=" + cid + "&offset=" + readOffset,
                new TypeReference<>() {});
        System.out.println(" ===>>> ack result: " + result);

        return result.getCode() == 1;
    }

    public Stat stat(String topic, String consumerId) {
        System.out.println(" ===>>> stat topic/cid: " + topic + "/" + consumerId);
        Result<Stat> result = HttpUtils.httpGet("http://" + brokerServer + "/stat?t=" + topic + "&cid=" + consumerId,
            new TypeReference<>() {});
        System.out.println(" ===>>> stat result: " + result);
        return result.getData();
    }

}
